package com.cennavi.funcs;

import java.util.Iterator;

import org.apache.spark.api.java.function.PairFunction;

import scala.Tuple2;

/** * @author  作者 :allen
    * @date 创建时间：2017年9月20日 上午8:51:28 
    * @version 1.0 
    */

public class FindMinFlowPairFunction implements PairFunction<Tuple2<String, Iterable<String>>, String, String> {

	private static final long serialVersionUID = 1L;
	
	private int algoType;

	public FindMinFlowPairFunction(int algoType) {
		this.algoType = algoType;
	}

	public Tuple2<String, String> call(Tuple2<String, Iterable<String>> arg0) throws Exception {
		String value = "";
		String returnValue = "";
		String[] data;
		int minFlow = 100000;
		String time = "";
		
		//为了适应三种算法
		if(this.algoType == 601){
			for(Iterator<String> iter = arg0._2.iterator();iter.hasNext();){
				value = (String)iter.next();
				data = value.split(",");
				if(Integer.parseInt(data[8]) < minFlow){
					minFlow = Integer.parseInt(data[8]);
					time = data[1];
					returnValue = String.valueOf(minFlow) + "," + time + "," + data[8] + ",null";
				}
			}
		}else if(this.algoType == 602){
			for(Iterator<String> iter = arg0._2.iterator();iter.hasNext();){
				value = (String)iter.next();
				data = value.split(",");
				if(Integer.parseInt(data[8]) < minFlow){
					minFlow = Integer.parseInt(data[8]);
					time = data[1];
					returnValue = String.valueOf(minFlow) + "," + time + ",null," + data[8];
				}
			}
		}else{
			for(Iterator<String> iter = arg0._2.iterator();iter.hasNext();){
				value = (String)iter.next();
				data = value.split(",");
				if(Integer.parseInt(data[8]) < minFlow){
					minFlow = Integer.parseInt(data[8]);
					time = data[1];
					returnValue = String.valueOf(minFlow) + "," + time + "," + data[11] + "," + data[12];
				}
			}
		}
		
		Tuple2<String, String> returnData = new Tuple2<String, String>(arg0._1, returnValue);
		return returnData;
	}

}
